package com.qingyun.material.utils;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;

import com.qingyun.material.dto.MaterialDTO;
import com.qingyun.material.entity.MaterialTask;
import com.qingyun.material.service.IMaterialService;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.qingyun.common.util.RedisConstants.*;


/**
 * Material专用 Redis Stream 消费者
 */
@Component
public class StreamMessageMaterialConsumer {

    private final RedisTemplate<String, Object> redisTemplate;
    private final IMaterialService materialService;

    // 多线程池
    private static final int CONSUMER_THREADS = Runtime.getRuntime().availableProcessors() << 1;
    private final ExecutorService consumerPool = Executors.newFixedThreadPool(CONSUMER_THREADS);

    public StreamMessageMaterialConsumer(RedisTemplate<String, Object> redisTemplate, IMaterialService materialService) {
        this.redisTemplate = redisTemplate;
        this.materialService = materialService;
        initConsumerGroup();
        startConsuming();
    }

    private void initConsumerGroup() {
        try {
            redisTemplate.opsForStream().createGroup(
                    MATERIAL_TASK_QUEUE,
                    ReadOffset.latest(),
                    MATERIAL_TASK_GROUP
            );
        } catch (Exception ignore) {}
    }
    //多线程消费逻辑：主线程拉消息，工作线程并行处理
    public void startConsuming() {
        //只用 1 条后台线程去不断拉取，防止连接竞争
        Executors.newSingleThreadExecutor().submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
                            Consumer.from(MATERIAL_TASK_GROUP, MATERIAL_TASK_CONSUMER),
                            StreamReadOptions.empty().block(Duration.ofSeconds(2)).count(10),
                            StreamOffset.create(MATERIAL_TASK_QUEUE, ReadOffset.lastConsumed())
                    );
                    if (records == null || records.isEmpty()) continue;

                    /* 一批消息分发给多条线程 */
                    CountDownLatch latch = new CountDownLatch(records.size());
                    for (MapRecord<String, Object, Object> r : records) {
                        consumerPool.execute(() -> {
                            try {
                                Map<Object, Object> map = r.getValue();
                                Object taskJson = map.get(FIELD_TASK_JSON);
                                if (taskJson != null) {
                                    JSONObject jsonObject = JSONUtil.parseObj(taskJson);
                                    MaterialTask materialTask = jsonObject.toBean(MaterialTask.class);
                                    MaterialDTO materialDTO= BeanUtil.copyProperties(materialTask,MaterialDTO.class);
                                    Long userId=materialTask.getUserId();
                                    materialService.saveMaterial(materialDTO,userId);
                                }
                                /* 业务成功后再 ack */
                                redisTemplate.opsForStream().acknowledge(MATERIAL_TASK_GROUP, r);
                            } catch (Exception ex) {
                                // 不 ack，消息进入 pending-list，后续可重试
                            } finally {
                                latch.countDown();
                            }
                        });
                    }
                    /* 主线程可继续去拉，不等待这批做完 */
                } catch (Exception e) {
                    try { Thread.sleep(500L); } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        });
    }
}